🤖 I'd like to double-check correctness when a data partition has more than one file group in the partitioned RLI. Looking at HoodieBackedTableMetadata.lookupIndexRecords for the dataTablePartition.isPresent() branch, it computes a single fileGroupIndex from only the first key (distinctSortedKeys.stream().findFirst().get()) and reads from fileSlicesForDataPartition.get(fileGroupIndex) — i.e. only one shard. Its comment says "all keys will be from the same shard index", which is true for the write path (SparkMetadataTableRecordLevelIndex.lookupRecords pre-partitions by file-group index before calling), but here we forward all recordKeys to that API per partition. If an IN (...) query has multiple keys that hash to different file groups inside a partition (which can happen once estimateFileGroupCount allocates more than 1 file group for a partition, bounded by record.level.index.max.filegroup.count = 10 by default), the lookup would only see the first key's shard, miss the others, and over-prune the candidate files → incorrect query results. The current tests use small data (≤5 records/partition) so each partition has just 1 file group and don't exercise this path. Could you confirm this, and either pre-bucket the keys by file group inside the loop (mirroring the write path) or add a test with RECORD_LEVEL_INDEX_MIN_FILE_GROUP_COUNT_PROP = 2+ and a multi-key IN filter to verify? @nsivabalan @yihua would appreciate your eyes on the assumption in lookupIndexRecords too.

_{- AI-generated; verify before applying. React 👍/👎 to flag quality.}

🤖 nit: could you make the key name self-documenting by including the data-skipping context? hoodie.datasource.read.record.index.max.partitions reads like a property of the index itself (its max partition count) rather than a fallback threshold for data skipping. Something like hoodie.datasource.read.record.index.data.skipping.max.partitions would match the field name RECORD_INDEX_MAX_DATA_SKIPPING_PARTITIONS and be unambiguous in a config file without the Javadoc nearby.

_{- AI-generated; verify before applying. React 👍/👎 to flag quality.}

🤖 nit: this branch uses the legacy/alternative key "hoodie.metadata.record.index.enable" as a bare string literal, while the if-branch above uses HoodieMetadataConfig.RECORD_LEVEL_INDEX_ENABLE_PROP.key. It's worth at least a brief comment explaining that this is the backward-compatible alternative key for GLOBAL_RECORD_LEVEL_INDEX_ENABLE_PROP — otherwise it looks like an accidental inconsistency with the key used in rliEnableOpts.

_{- AI-generated; verify before applying. React 👍/👎 to flag quality.}

is this the num for per query task?

🤖 Could you double-check what happens if a candidate partition exists in the data table but isn't yet present in getBucketizedFileGroupsForPartitionedRLI (e.g. during an RLI backfill, or any transient skew)? The .filter { partitions.contains(partition) } silently drops such partitions; the lookup then never queries them, and filterCandidateFiles ends up filtering all their files out of the candidate set — so a matching record there would be skipped at read time. The Flink path (hudi-flink/.../RecordLevelIndex.java#groupKeysByPartitionAndShard) explicitly checks-state here and refuses to prune; should the Spark path behave the same (fall back to Option.empty if any candidate partition is missing from the indexed map) instead of silently over-pruning? @yihua

_{- AI-generated; verify before applying. React 👍/👎 to flag quality.}

🤖 The PR description says: 'Wire HoodieFileIndex to instantiate record-index support through the new factory and read the new hoodie.datasource.read.record.index.max.partitions threshold option.' — but in the current revision this is a hard-coded private[hudi] val MAX_PARTITIONS = 10, and I can't find any read of that option anywhere in the source tree. Was the config dropped intentionally, or did the wiring get lost in a rebase? With Flink's analogous threshold set to 3 and Spark to 10 hard-coded, making this user-tunable seems valuable for tables with many partitions that still want RLI pruning.

_{- AI-generated; verify before applying. React 👍/👎 to flag quality.}

🤖 nit: the three-arg Stream.collect() mutable-reduction form is a bit surprising here — could you use collect(Collectors.toMap(Pair::getKey, Pair::getValue)) instead? It makes the intent immediately obvious.

_{- AI-generated; verify before applying. React 👍/👎 to flag quality.}

🤖 nit: using foldLeft to build a mutable Map mixes functional and imperative styles in a way that can trip up readers — have you considered simplifying to .map(t => t._2.getFileId -> t._2.getPartitionPath).toMap? That also lets lookupRecordKeys return Option[Map[String, String]] (immutable) since the map is never mutated after construction.

_{- AI-generated; verify before applying. React 👍/👎 to flag quality.}